package com.design.cfr;

import com.design.cfr.model.ChainResponsibilityElementResult;
import com.design.cfr.model.ChainResponsibilityResult;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * Normal chain responsibility abstract
 *
 * @description: 通用责任连，执行方式为异步执行。必须确保责任链头部继承SyncChainResponsibilityAbstract，
 * 即用继承SyncChainResponsibilityAbstract的类启动
 * 分片执行不会呗异常终止
 * T --> apply(T t)的参数
 * @author: cyj
 * @date 2022 -08-12 15:38:51
 */
@Slf4j
public abstract class SyncChainResponsibilityAbstract<T>
        extends ChainResponsibility<T, ChainResponsibilityElementResult, ChainResponsibilityResult> {

    private static final Executor executor = new ThreadPoolExecutor(
            1,
            20,
            10L,
            TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>(),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    @Override
    public ChainResponsibilityResult start(T t) {
        return startNothrow(t);
    }

    @Override
    public ChainResponsibilityResult startNothrow(T t){
        return startCommon(t, (ChainResponsibility<T, ChainResponsibilityElementResult, ChainResponsibilityResult> element) -> {
            ChainResponsibilityElementResult apply;
            try {
                apply = element.apply(t);
            } catch (Exception e) {
                log.warn("执行责任链异常!", e);
                apply = ChainResponsibilityElementResult.builder()
                        .pass(false)
                        .aClass(element.getClass())
                        .message(e.getMessage())
                        .build();
            }
            return apply;
        });
    }

    private ChainResponsibilityResult startCommon(T t,
                                                  Function<ChainResponsibility<T, ChainResponsibilityElementResult, ChainResponsibilityResult>,
                                                          ChainResponsibilityElementResult> resultSupplier) {
        // new结果对象
        ChainResponsibilityResult chainResponsibilityResult = new ChainResponsibilityResult();
        List<ChainResponsibility<T, ChainResponsibilityElementResult, ChainResponsibilityResult>> chainResponsibilities = super.toList();

        // 组装回调线程
        List<CompletableFuture<ChainResponsibilityElementResult>> completableFutures = chainResponsibilities
                .stream()
                .map(
                        chainResponsibility ->
                                CompletableFuture.supplyAsync(
                                        () -> resultSupplier.apply(chainResponsibility)))
                .collect(Collectors.toList());

        // 聚合线程执行结果
        CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]));


        completableFutures.stream().forEach(completableFuture -> {
            try {
                ChainResponsibilityElementResult chainResponsibilityElementResult = completableFuture.get();
                chainResponsibilityResult.getApplyResultElements().add(chainResponsibilityElementResult);
            } catch (InterruptedException | ExecutionException e) {
                log.warn("执行责任链异常, 线程执行异常!", e);
                chainResponsibilityResult.getApplyResultElements().add(ChainResponsibilityElementResult.builder()
                        .pass(false)
                        .aClass(completableFuture.getClass())
                        .message(e.getMessage())
                        .build()
                );
            }
        });
        // 设置结果
        chainResponsibilityResult.setResult();
        return chainResponsibilityResult;
    }
}
